package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.VisitorStates;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import static com.atguigu.gmall.realtime.common.Constant.*;

import java.time.Duration;
import java.util.Map;

/**
 * DWS 用户主题宽表
 *
 * @author Archie
 * @date 2021-10-27 8:38
 * @description
 */
public class DwsVisitorStatesApp extends BaseAppV2 {

	public static void main(String[] args) {
		new DwsVisitorStatesApp().init(
				4001,
				1,
				"DwsVisitorStatesApp",
				"DwsVisitorStatesApp",
				TOPIC_DWD_PAGE, TOPIC_DWM_UV, TOPIC_DWM_USER_JUMP_DETAIL
		);
	}

	@Override
	protected void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> streams) {
		// 1. 解析并Union成 VisitorStates实体
		DataStream<VisitorStates> vsDataStream = parseAndUnionOne(streams);
		// 2. 分组开窗聚合
		SingleOutputStreamOperator<VisitorStates> aggregateStream = aggregate(vsDataStream);
		// 3. 写入DWS层
		writeToClickHouse(aggregateStream);
	}

	private void writeToClickHouse(SingleOutputStreamOperator<VisitorStates> aggregateStream) {
		aggregateStream.addSink(FlinkSinkUtil.getClickHouseSink("gmall2021", "visitor_stats_2021", VisitorStates.class));
	}

	private SingleOutputStreamOperator<VisitorStates> aggregate(DataStream<VisitorStates> vsDataStream) {
		return vsDataStream
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
								.<VisitorStates>forBoundedOutOfOrderness(Duration.ofSeconds(20))
								.withTimestampAssigner((obj, ts) -> obj.getTs())
				)
				.keyBy(vs -> vs.getVc() + "_" + vs.getCh() + "_" + vs.getAr() + "_" + vs.getIs_new())
				.window(TumblingEventTimeWindows.of(Time.seconds(5)))
				.sideOutputLateData(new OutputTag<VisitorStates>("late"){}) // 用于鉴定是否有数据晚到
				.reduce(
						new ReduceFunction<VisitorStates>() {
							/**
							 * 聚合逻辑
							 * @param vs1 总叠加数据
							 * @param vs2 被叠加数据
							 * @return
							 * @throws Exception
							 */
							@Override
							public VisitorStates reduce(VisitorStates vs1, VisitorStates vs2) throws Exception {
								vs1.setPv_ct(vs1.getPv_ct() + vs2.getPv_ct());
								vs1.setUv_ct(vs1.getUv_ct() + vs2.getUv_ct());
								vs1.setSv_ct(vs1.getSv_ct() + vs2.getSv_ct());
								vs1.setUj_ct(vs1.getUj_ct() + vs2.getUj_ct());
								vs1.setDur_sum(vs1.getDur_sum() + vs2.getDur_sum());

								return vs1;
							}
						},
						new ProcessWindowFunction<VisitorStates, VisitorStates, String, TimeWindow>() {
							/**
							 * 深入抓取窗口信息（开启统计时间&结束统计时间）
							 * @param s 键
							 * @param context 上下文
							 * @param elements 看似迭代，此处仅有一条数据
							 * @param out 收集器
							 * @throws Exception
							 */
							@Override
							public void process(String s, Context context, Iterable<VisitorStates> elements, Collector<VisitorStates> out) throws Exception {
								VisitorStates vs = elements.iterator().next();
								TimeWindow window = context.window();
								String stt = AtguiguUtil.toDataTimeString(window.getEnd());
								String edt = AtguiguUtil.toDataTimeString(window.getStart());
								vs.setStt(stt);
								vs.setEdt(edt);
								vs.setTs(System.currentTimeMillis()); // 更新统计时间为当前系统时间（可选）
								out.collect(vs);
							}
						}
				);
	}

	private DataStream<VisitorStates> parseAndUnionOne(Map<String, DataStreamSource<String>> streams) {
		// 1. pv 和 持续访问时间
		SingleOutputStreamOperator<VisitorStates> pvAndDurSumStream = streams
				.get(TOPIC_DWD_PAGE)
				.map(json -> {
					JSONObject obj = JSON.parseObject(json);

					JSONObject common = obj.getJSONObject("common");
					String vc = common.getString("vc");
					String ch = common.getString("ch");
					String ar = common.getString("ar");
					String is_new = common.getString("is_new");

					Long ts = obj.getLong("ts");

					Long dur_sum = obj.getJSONObject("page").getLong("during_time");

					return new VisitorStates(
							"", "",
							vc, ch, ar, is_new,
							0L, 1L, 0L, 0L, dur_sum,
							ts);

				});

		// 2. uv
		SingleOutputStreamOperator<VisitorStates> uvStream = streams
				.get(TOPIC_DWM_UV)
				.map(json -> {
					JSONObject obj = JSON.parseObject(json);

					JSONObject common = obj.getJSONObject("common");
					String vc = common.getString("vc");
					String ch = common.getString("ch");
					String ar = common.getString("ar");
					String is_new = common.getString("is_new");

					Long ts = obj.getLong("ts");

					return new VisitorStates(
							"", "",
							vc, ch, ar, is_new,
							1L, 0L, 0L, 0L, 0L,
							ts);

				});
		// 3. 跳出次数
		SingleOutputStreamOperator<VisitorStates> ujStream = streams
				.get(TOPIC_DWM_USER_JUMP_DETAIL)
				.map(json -> {
					JSONObject obj = JSON.parseObject(json);

					JSONObject common = obj.getJSONObject("common");
					String vc = common.getString("vc");
					String ch = common.getString("ch");
					String ar = common.getString("ar");
					String is_new = common.getString("is_new");

					Long ts = obj.getLong("ts");

					return new VisitorStates(
							"", "",
							vc, ch, ar, is_new,
							0L, 0L, 0L, 1L, 0L,
							ts);

				});

		// 4. sv
		SingleOutputStreamOperator<VisitorStates> svStream = streams
				.get(TOPIC_DWD_PAGE)
				.flatMap(new FlatMapFunction<String, VisitorStates>() {
					@Override
					public void flatMap(String json, Collector<VisitorStates> out) throws Exception {
						JSONObject obj = JSON.parseObject(json);

						String lastPageId = obj.getJSONObject("page").getString("last_page_id");
						if (lastPageId == null || lastPageId.length() == 0) {
							JSONObject common = obj.getJSONObject("common");
							String vc = common.getString("vc");
							String ch = common.getString("ch");
							String ar = common.getString("ar");
							String is_new = common.getString("is_new");

							Long ts = obj.getLong("ts");

							VisitorStates vs = new VisitorStates(
									"", "",
									vc, ch, ar, is_new,
									0L, 0L, 1L, 0L, 0L,
									ts);
							out.collect(vs);

						}

					}
				});

		return pvAndDurSumStream.union(
				uvStream,
				ujStream,
				svStream
		);

	}
}
